#include "TcpConnection.h"
#include "Channel.h"
#include "EventLoop.h"
#include "Logger.h"
#include "Socket.h"
#include <errno.h>
#include <functional>
#include <memory>

extern EventLoop *check_loop_not_null(EventLoop *loop);

TcpConnection::TcpConnection(EventLoop *loop, const std::string &name,
                             int sockfd, const InetAddress &localAddr,
                             const InetAddress &peerAddr)
    : loop_(check_loop_not_null(loop)), name_(name), state_(CONNECTING),
      reading_(true), socket_(new Socket(sockfd)),
      channel_(new Channel(loop, sockfd)), localAddr_(localAddr),
      peerAddr_(peerAddr), highWaterMark_(64 * 1024 * 1024)
{
    channel_->setReadCallback(
        std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
    channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
    channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
    channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));
    LOG_INFO("TcpConnection::ctor[%s] at fd=%d", name_.c_str(), socket_->fd());
    socket_->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
    LOG_INFO("TcpConnection::dtor[%s] at fd=%d state=%d", name_.c_str(),
             socket_->fd(), (int)state_);
}

void TcpConnection::handleRead(Timestamp receiveTime)
{

    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    LOG_DEBUG("read fd=%d =%d", channel_->fd(), n);
    if (n > 0)
    {
        // 已建立连接的用户，有可读事件发生了，调用用户传入的回调操作
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
}
void TcpConnection::handleWrite()
{
    LOG_DEBUG("write fd=%d state=%d", channel_->fd(), (int)state_);
    if (channel_->isWriting())
    {
        ssize_t n = ::write(channel_->fd(), outputBuffer_.peek(),
                            outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->disableWriting();
                if (writeCompleteCallback_)
                {
                    loop_->queueInLoop(
                        std::bind(writeCompleteCallback_, shared_from_this()));
                }
            }
            if (state_ == DISCONNECTING)
            {
                shutdownInLoop();
            }
        }
        else
        {
            LOG_ERROR("TcpConnection::handleWrite");
        }
    }
    else
    {
        LOG_ERROR("TcpConnection fd =%d is down, no more writing",
                  channel_->fd());
    }
}
void TcpConnection::handleClose()
{
    LOG_INFO("close fd=%d state=%d", channel_->fd(), (int)state_);

    setState(DISCONNECTED);
    LOG_DEBUG("state=%d", (int)state_);
    channel_->disableAll();

    TcpConnectionPtr guardThis(shared_from_this());

    connectionCallback_(guardThis);
    closeCallback_(guardThis);
}
void TcpConnection::handleError()
{
    int optval;
    socklen_t optlen = sizeof(optval);
    int err = 0;
    if (::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) <
        0)
    {
        err = errno;
    }
    else
    {
        err = optval;
    }
    LOG_ERROR("name %s - SOL_ERROR %d", name_.c_str(), err);
}

// 应用写数据快，而内核发送慢，需要把待发送数据写入缓冲区，而且设置水位回调
void TcpConnection::sendInLoop(const void *data, size_t len)
{
    ssize_t nwrote = 0;
    size_t remaining = len;

    bool faultError = false;

    // 之前调用过该connection的shutdown，不能再进行发送了
    if (state_ == DISCONNECTED)
    {
        LOG_ERROR("disconnected, give up writing");
        return;
    }

    // 表示channel_第一次开始写数据，而且缓冲区没有待发送数据
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        nwrote = ::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeCompleteCallback_)
            {
                // 数据一次性发送完成，就不用再给channel设置EPOLLOUT事件
                loop_->queueInLoop(
                    std::bind(writeCompleteCallback_, shared_from_this()));
            }
        }
        else
        {
            nwrote = 0;
            if (errno != EAGAIN)
            {
                LOG_ERROR("send error");
                if (errno == EPIPE || errno == ECONNRESET)
                {
                    faultError = true;
                }
            }
        }
    }

    // 当前这次write没有把所有数据发送出去，剩余数据需要保存，然后给channel注册epollout事件，poller发现
    // tcp的发送缓冲区有空间时，会通知相应的channel调用writeCallback_回调方法，也就是TcpConnection::handWrite方法
    // 把剩余数据发送完成
    if (!faultError && remaining > 0)
    {
        // 目前发送缓冲区剩余的待发送数据
        size_t oldLen = outputBuffer_.readableBytes();

        if (oldLen + remaining >= highWaterMark_ && oldLen < highWaterMark_)
        {
            if (highWaterMark_)
            {
                loop_->queueInLoop(std::bind(highWaterMarkCallback_,
                                             shared_from_this(),
                                             oldLen + remaining));
            }
            outputBuffer_.append((char *)data + nwrote, remaining);
            if (!channel_->isWriting()) // 注册写事件，让poller通知channel写操作
            {
                channel_->enableWriting();
            }
        }
    }
}
void TcpConnection::shutdown()
{
    if (state_ == CONNECTED)
    {
        setState(DISCONNECTING);
        loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
}
void TcpConnection::shutdownInLoop()
{
    if (!channel_->isWriting()) // 说明当前outputBuffer_中的数据已经全部发送完成
    {
        socket_->shutdownWrite();
    }
}

void TcpConnection::send(const std::string &msg)
{
    if (state_ == CONNECTED)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(msg.c_str(), msg.size());
        }
        else
        {
            loop_->runInLoop(std::bind(&TcpConnection::sendInLoop, this,
                                       msg.c_str(), msg.size()));
        }
    }
}

void TcpConnection::connectEstablished()
{
    setState(CONNECTED);
    channel_->tie(shared_from_this());
    channel_->enableReading(); // channel_向poller注册epollin事件

    // 新连接建立
    connectionCallback_(shared_from_this());
}
void TcpConnection::connectDestroyed()
{
    if (state_ == CONNECTED)
    {
        setState(DISCONNECTED);
        channel_->disableAll(); // 把channel_所有感兴趣的事件从poller中删掉
    }
    LOG_DEBUG("remove channel_->fd %d", channel_->fd());
    channel_->remove(); // 把channel从poller中删掉
}
